package cloudbus

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"zstackgo/java"
	"zstackgo/pkg/core/componentloader"
	"zstackgo/pkg/core/thread"
	"zstackgo/pkg/core/timeout"
	"zstackgo/pkg/header"
	"zstackgo/pkg/header/core"
	"zstackgo/pkg/header/errorcode"
	"zstackgo/pkg/header/message"
	"zstackgo/pkg/header/rest"
	"zstackgo/spring"
	"zstackgo/todo"

	mhttp "github.com/meilihao/golib/v2/http"
	"github.com/meilihao/golib/v2/log"
	"go.uber.org/zap"
)

var (
	_ CloudBus = new(CloudBusImpl3)
)

const (
	NO_NEED_REPLY_MSG    = "noReply"
	CORRELATION_ID       = "correlationId"
	REPLY_TO             = "replyTo"
	IS_MESSAGE_REPLY     = "isReply"
	SERVICE_ID_SPLITTER  = ":::"
	THREAD_CONTEXT_STACK = "thread-context-stack"
	THREAD_CONTEXT       = "thread-context"
	TASK_CONTEXT         = "task-context"
	HTTP_BASE_URL        = "/cloudbus"
)

var (
	cloudBusService *CloudBusImpl3

	SERVICE_ID string
	EVENT_ID   string

	services                                = []header.Service{}
	replyMessageMarshaller                  = sync.Map{}             // Map<Class, List<MarshalReplyMessageExtensionPoint>>
	apiExts                                 []atomic.Value           // List<RestAPIExtensionPoint>
	msgExts                                 []CloudBusExtensionPoint // List<CloudBusExtensionPoint>
	beforeDeliveryMessageInterceptors       = map[string][]message.BeforeDeliveryMessageInterceptor{}
	beforeSendMessageInterceptors           = map[string][]message.BeforeSendMessageInterceptor{}
	beforeEventPublishInterceptors          = map[string][]message.BeforePublishEventInterceptor{}
	beforeDeliveryMessageInterceptorsForAll = []message.BeforeDeliveryMessageInterceptor{}
	beforeSendMessageInterceptorsForAll     = []message.BeforeSendMessageInterceptor{}
	beforeEventPublishInterceptorsForAll    = []message.BeforePublishEventInterceptor{}
	eventListeners                          = map[string]map[string]CloudBusEventListener{}

	filterMsgNames   = map[string]bool{}          //  Set<String>
	endPoints        = map[string]EndPoint{}      // Map<String, EndPoint>
	envelopes        = sync.Map{}                 // Map<String, Envelope>
	messageConsumers = map[string]java.Consumer{} // Map<String, java.util.function.Consumer>
	lock             = sync.RWMutex{}
)

func init() {
	cloudBusService = &CloudBusImpl3{}
	cloudBusService.messageConsumer = &messageConsumer{
		_bus: cloudBusService,
	}
	spring.Register(cloudBusService, "CloudBus", nil)

	if StaticCloudBusGlobalProperty.MessageLog != "" {
		msgNames := strings.Split(StaticCloudBusGlobalProperty.MessageLog, ",")
		for _, name := range msgNames {
			filterMsgNames[name] = true

		}
	}
	if StaticCloudBusGlobalProperty.UnitTestOn && false { // todo && !Platform.isSimulatorOn())
		StaticCloudBusGlobalProperty.HttpContextPath = ""
		StaticCloudBusGlobalProperty.HttpPort = 8989
	}
}

type CloudBusImpl3 struct {
	Thdf               thread.ThreadFacadeImpl        `autowire:"ThreadFacadeImpl"`
	TimeoutMgr         timeout.ApiTimeoutManager      `autowire:"ApiTimeoutManagerImpl"`
	DestMaker          ResourceDestinationMaker       `autowire:"ResourceDestinationMakerImpl"`
	PluginRgty         componentloader.PluginRegistry `autowire:"PluginRegistry"`
	DeadMessageManager `autowire:"DeadMessageManager"`
	messageConsumer    java.Consumer
}

type Envelope interface {
	Count(msg message.Message)
	Ack(reply *message.MessageReply)
	Cancel(err string)
	Timeout()
}

var (
	_ Envelope = new(EnvelopeImpl)
)

type EnvelopeImpl struct {
	msg       message.NeedReplyMessage
	called    atomic.Bool // 判断这个msg是否已经发出去
	StartTime time.Time
	envelopes sync.Map
	callBack  CloudBusCallBack
}

func NewEnvelopeImpl() *EnvelopeImpl {
	i := new(EnvelopeImpl)

	if StaticCloudBusGlobalConfig.STATISTICS_ON {
		i.StartTime = time.Now()
	}

	return i
}

func (impl *EnvelopeImpl) Count(msg message.Message) {

}

// msg 发送成功时候调用这个方法
func (impl *EnvelopeImpl) Ack(reply *message.MessageReply) {
	// 计算该msg耗时
	impl.Count(impl.msg)

	impl.envelopes.Delete(impl.msg.GetId())

	// 如果更新失败，说明这个消息已经被发送过了
	if !impl.called.CompareAndSwap(false, true) {
		return
	}

	// todo timeoutTaskReceipt.cancel(); // 取消一个计算超时的任务. thdf.submitTimeoutTask: 计算超时，往线程池提交一个任务

	// 调用注册的callback
	impl.callBack.Run(reply)
}

func (impl *EnvelopeImpl) Cancel(err string) {
	impl.envelopes.Delete(impl.msg.GetId())

	if !impl.called.CompareAndSwap(false, true) {
		return
	}

	// todo timeoutTaskReceipt.cancel();

	impl.callBack.Run(createErrorReply(impl.msg, &errorcode.ErrorCode{
		Code:    errorcode.CANCEL_SysErrors,
		Details: err,
	}))
}

// 消息超时时调用的逻辑
func (impl *EnvelopeImpl) Timeout() {
	impl.envelopes.Delete(impl.msg.GetId())

	// 已经被调用过则返回
	if !impl.called.CompareAndSwap(false, true) {
		return
	}

	// todo timeoutTaskReceipt.cancel();

	impl.callBack.Run(createTimeoutReply(impl.msg))
}

type ConsumerReceipt interface {
	Cancel()
}

type consumerReceiptImpl func()

func (f consumerReceiptImpl) Cancel() {
	f()
}

func (impl *CloudBusImpl3) on(serviceId string, consumer java.Consumer) ConsumerReceipt {
	var old java.Consumer
	lock.RLock()
	old = messageConsumers[serviceId]
	lock.RUnlock()

	if old != nil && old != consumer {
		panic(fmt.Sprintf("duplicate Consumer[%v,%v] for the same service id[%s]", old.GetClass(), consumer.GetClass(), serviceId))
	}

	lock.Lock()
	defer lock.Unlock()

	messageConsumers[serviceId] = consumer

	return consumerReceiptImpl(func() {
		lock.Lock()
		defer lock.Unlock()

		delete(messageConsumers, serviceId)
	})
}

type eventConsumer struct {
}

func (er *eventConsumer) GetClass() string {
	return "eventConsumer"
}

func (er *eventConsumer) callListener(e message.Event, listener CloudBusEventListener) {
	listener.HandleEvent(e)
}

func (er *eventConsumer) Accept(t interface{}) {
	evt := t.(message.Event)
	log.Glog.Debug(fmt.Sprintf("[event received]: %s", dumpMessage(evt)))

	lock.RLock()
	defer lock.RUnlock()
	ls, _ := eventListeners[evt.GetType().String()]
	if ls == nil {
		return
	}

	for _, l := range ls {
		er.callListener(evt, l)
	}
}

type messageConsumer struct {
	_bus *CloudBusImpl3
}

func (er *messageConsumer) GetClass() string {
	return "messageConsumer"
}

func (er *messageConsumer) Accept(t interface{}) {
	msg := t.(message.Message)

	if IsLogMessage(msg) {
		log.Glog.Debug(fmt.Sprintf("[msg received]: %s", dumpMessage(msg)))
	}

	r, ok := msg.(*message.MessageReply)
	if ok {
		er._bus.beforeDeliverMessage(msg)

		correlationId := r.GetHeaderEntry(CORRELATION_ID)
		eTmp, _ := envelopes.Load(correlationId)
		if eTmp == nil {
			log.Glog.Warn(fmt.Sprintf("received a message reply but no envelope found,"+
				"maybe the message request has been timeout or sender doesn't care about reply."+
				"drop it. reply dump:\n%s", dumpMessage(r)))
			return
		}
		e := eTmp.(Envelope)
		e.Ack(r)
	} else {
		er._bus.DealWithUnknowMessage(msg)
	}
}

func dumpMessage(msg message.Message) string {
	return fmt.Sprintf("%s %s", msg.GetClass(), todo.ToJson(msg))
}

type EndPoint interface {
	Active()
	Inactive()
}

func (impl *CloudBusImpl3) ActiveService(serv header.Service) {
	impl.ActiveServiceById(serv.GetId())
}

func (impl *CloudBusImpl3) ActiveServiceById(id string) {
	e := endPoints[id]
	e.Active()
}

func (impl *CloudBusImpl3) DeActiveService(serv header.Service) {
	impl.DeActiveServiceById(serv.GetId())
}

func (impl *CloudBusImpl3) DeActiveServiceById(id string) {
	e := endPoints[id]
	e.Inactive()
}

func (impl *CloudBusImpl3) Send(msg message.Message) {
	impl.send(msg, true)
}

func (impl *CloudBusImpl3) Sends(msgs []message.Message) {
	for _, m := range msgs {
		impl.Send(m)
	}
}

func createErrorReply(m message.NeedReplyMessage, er *errorcode.ErrorCode) *message.MessageReply {
	r := &message.MessageReply{}
	r.PutHeaderEntry(CORRELATION_ID, m.GetId())
	r.SetError(er)
	r.SetServiceId(m.GetServiceId())
	return r
}

func createTimeoutReply(m message.NeedReplyMessage) *message.MessageReply {
	return createErrorReply(m, &errorcode.ErrorCode{
		Code:    errorcode.TIMEOUT_SysErrors,
		Details: m.Error(),
	})
}

func (impl *CloudBusImpl3) SendNeedReply(msg message.NeedReplyMessage, callback CloudBusCallBack) {
	// todo: evaluateMessageTimeout(msg);
	if msg.GetTimeout() <= 1 {
		callback.Run(createTimeoutReply(msg))
		return
	}

	e := EnvelopeImpl{
		msg:       msg,
		envelopes: envelopes,
		callBack:  callback,
	}
	envelopes.Store(msg.GetId(), e)
	for _, m := range msgExts {
		m.AfterAddEnvelopes(msg.GetId())
	}
	impl.send(msg, false)
}

func (impl *CloudBusImpl3) SendNeedReplies(msgs []message.NeedReplyMessage, callBack CloudBusListCallBack) {
	impl.SendNeedRepliesWithParallel(msgs, len(msgs), callBack)
}

func (impl *CloudBusImpl3) SendNeedRepliesWithParallel(msgs []message.NeedReplyMessage, parallelLevel int, callBack CloudBusListCallBack) {
	if len(msgs) == 0 {
		panic("you cannot pass an empty message list to me")
	}

	ch := make(chan struct{}, parallelLevel)
	wg := sync.WaitGroup{}
	replies := sync.Map{}

	for i := range msgs {
		ch <- struct{}{}
		wg.Add(1)

		go func(i int) {
			defer wg.Done()

			t := sync.WaitGroup{}
			cb := NewCloudBusCallBackImpl[sync.Map](func(reply *message.MessageReply) {
				replies.Store(msgs[i].GetId(), reply)
				t.Done()
			}, replies)

			impl.SendNeedReply(msgs[i], cb)
			t.Wait()

			<-ch
		}(i)
	}

	wg.Wait()

	results := make([]*message.MessageReply, 0, len(msgs))

	var value any
	var ok bool
	for _, v := range msgs {
		value, ok = replies.Load(v.GetId())
		if !ok {
			panic(fmt.Errorf("missage mesage: %s", v.GetId()))
		}

		results = append(results, value.(*message.MessageReply))
	}

	callBack.Run(results)
}

func (impl *CloudBusImpl3) Routes(msgs []message.Message) {
	for _, msg := range msgs {
		impl.Route(msg)
	}
}

func (impl *CloudBusImpl3) Route(msg message.Message) {
	if msg.GetServiceId() == "" {
		panic(fmt.Sprintf("service id cannot be null: %s", msg.GetClass()))
	}

	if v, ok := msg.(message.NeedReplyMessage); ok {
		_ = v
		// todo :evaluateMessageTimeout((NeedReplyMessage) msg);
	}

	impl.doSendAndCallExtensions(msg)
}

func (impl *CloudBusImpl3) callReplyBeforeDropExtensions(msg message.Message, msgReq message.NeedReplyMessage) {
	extsTmp, ok := replyMessageMarshaller.Load(msg.GetClass())
	if ok {
		exts := extsTmp.([]MarshalReplyMessageExtensionPoint)

		for _, ext := range exts {
			ext.MarshalReplyMessageBeforeDropping(msg, msgReq)
		}
	}
}

func (impl *CloudBusImpl3) callReplyPreSendingExtensions(msg message.Message, msgReq message.NeedReplyMessage) {
	extsTmp, ok := replyMessageMarshaller.Load(msg.GetClass())
	if ok {
		exts := extsTmp.([]MarshalReplyMessageExtensionPoint)

		for _, ext := range exts {
			ext.MarshalReplyMessageBeforeSending(msg, msgReq)
		}
	}

	for _, v := range apiExts {
		e := v.Load().(rest.RestAPIExtensionPoint)
		e.BeforeAPIResponse(msg)
	}
}

func (impl *CloudBusImpl3) Reply(request message.Message, reply *message.MessageReply) {
	if request.GetHeaderEntry(NO_NEED_REPLY_MSG) == "true" {
		if v, ok := request.(message.NeedReplyMessage); ok {
			impl.callReplyBeforeDropExtensions(reply, v)
		}

		log.Glog.Debug(fmt.Sprintf("%s in message%s is set, drop reply%s", NO_NEED_REPLY_MSG,
			dumpMessage(request), dumpMessage(reply)))

		return
	}

	reply.PutHeaderEntry(IS_MESSAGE_REPLY, "true")
	reply.PutHeaderEntry(CORRELATION_ID, request.GetId())
	reply.SetServiceId(request.GetHeaderEntry(REPLY_TO))

	if v, ok := request.(message.NeedReplyMessage); ok {
		impl.callReplyPreSendingExtensions(reply, v)
	}

	impl.doSend(reply)
}

func (impl *CloudBusImpl3) Cancel(correlationId string, err string) {
	e, ok := envelopes.Load(correlationId)
	if !ok {
		log.Glog.Warn(fmt.Sprintf("received a correlationId[%s] but no envelope found,", correlationId))
		return
	}

	tmp := e.(*EnvelopeImpl)
	tmp.Cancel(err)
}

func (impl *CloudBusImpl3) Publishs(events []message.Event) {
	for _, v := range events {
		impl.Publish(v)
	}
}

func (impl *CloudBusImpl3) Publish(event message.Event) {
	if aevt, ok := event.(*message.APIEvent); ok {
		if aevt.ApiId == "" {
			panic(fmt.Errorf("apiId of %s cannot be null", aevt.GetClass()))
		}
	}

	impl.callReplyPreSendingExtensions(event, nil)

	var is []message.BeforePublishEventInterceptor

	lock.RLock()
	is = beforeEventPublishInterceptors[event.GetClass()]
	lock.RUnlock()

	for _, i := range is {
		i.BeforePublishEvent(event)
	}
	for _, i := range beforeEventPublishInterceptorsForAll {
		i.BeforePublishEvent(event)
	}

	impl.doPublish(event)
}

type MessageSender struct {
	_bus             *CloudBusImpl3
	msg              message.Message
	managementNodeId string
	serviceId        string
	_localSend       bool
}

func NewMessageSender(msg message.Message) *MessageSender {
	s := &MessageSender{
		_bus: cloudBusService,
		msg:  msg,
	}

	if _, ok := msg.(message.Event); ok {
		s.serviceId = EVENT_ID
	} else {
		s.serviceId = msg.GetServiceId()
	}

	ids := strings.SplitAfterN(s.serviceId, SERVICE_ID_SPLITTER, 2)
	if len(ids) == 1 {
		s.managementNodeId = PlatformGetManagementServerId()
	} else {
		s.managementNodeId = ids[0]
	}

	s._localSend = !StaticCloudBusGlobalProperty.HttpAlways && s.managementNodeId == PlatformGetManagementServerId()

	return s
}

func PlatformGetManagementServerId() string {
	return ""
}

func (s *MessageSender) Send() {
	if err := s.doSend(); err != nil {
		s.replyErrorIfNeeded(&errorcode.ErrorCode{
			Code:    errorcode.OPERATION_SysErrors,
			Details: err.Error(),
		})
	}
}

func (s *MessageSender) doSend() error {
	if _, ok := s.msg.(message.Event); ok {
		return s.eventSend()
	}

	if s._localSend {
		return s.localSend()
	} else {
		return s.httpSend()
	}
}

func (s *MessageSender) httpSendInQueue(ip string) error {
	return s.httpSendByIp(ip)
}

func (s *MessageSender) httpSend() error {
	s.buildSchema(s.msg)

	ni := s._bus.DestMaker.GetNodeInfo(s.managementNodeId)
	if ni == nil { // ManagementNodeNotFound
		needLog := true
		if v, ok := s.msg.(*message.MessageReply); ok {
			needLog = !s._bus.DeadMessageManager.HandleManagementNodeNotFoundError(s.managementNodeId, v, java.RunnableImpl(func() {
				ip := s._bus.DestMaker.GetNodeInfo(s.managementNodeId).NodeIp // todo: may be bad code, 还是会找不到node
				s.httpSendInQueue(ip)
			}))
		}
		if needLog {
			log.Glog.Error("no managementNodeId to send", zap.String("managementNodeId", s.managementNodeId), zap.String("msg", dumpMessage(s.msg)))
		}
	} else {
		return s.httpSendInQueue(ni.NodeIp)
	}

	return nil
}

func (s *MessageSender) httpSendByIp(ip string) error {
	var url string
	if StaticCloudBusGlobalProperty.HttpContextPath == "" {
		url = fmt.Sprintf("http://%s:%s%s", ip, StaticCloudBusGlobalProperty.HttpPort, HTTP_BASE_URL)
	} else {
		url = fmt.Sprintf("http://%s:%s/%s/%s", ip, StaticCloudBusGlobalProperty.HttpPort, StaticCloudBusGlobalProperty.HttpContextPath, HTTP_BASE_URL)
	}

	data, _ := json.Marshal(s.msg)

	opt := &mhttp.Option{
		Method:     http.MethodPost,
		Path:       url,
		Body:       bytes.NewReader(data),
		RetryCount: 2,
	}
	resp, err := mhttp.Do(context.TODO(), opt)
	if err != nil {
		s.replyErrorIfNeeded(&errorcode.ErrorCode{
			Code:    errorcode.OPERATION_SysErrors,
			Details: err.Error(),
		})
	}
	if !(resp.Code >= 200 && resp.Code < 300) {
		s.replyErrorIfNeeded(&errorcode.ErrorCode{
			Code:    errorcode.OPERATION_SysErrors,
			Details: fmt.Sprintf("HTTP ERROR, status code: %d, body: %s", resp.Code, string(resp.Body)),
		})
	}

	return nil
}

func (s *MessageSender) replyErrorIfNeeded(er *errorcode.ErrorCode) *errorcode.ErrorCode {
	if _, ok := s.msg.(message.APIMessage); ok {
		return er
	}
	if v, ok := s.msg.(message.NeedReplyMessage); ok {
		reply := createErrorReply(v, er)
		s._bus.messageConsumer.Accept(reply)
	}

	return nil
}

func (s *MessageSender) buildSchema(msg message.Message) error {
	msg.PutHeaderEntry(HEADER_SCHEMA, msg.GetClass())

	return nil
}

func (s *MessageSender) eventSend() error {
	s.buildSchema(s.msg)
	s.localSend()

	nodes := s._bus.DestMaker.GetAllNodeInfo()
	for _, node := range nodes {
		if node.NodeUuid != PlatformGetManagementServerId() { // not current node
			s.httpSendInQueue(node.NodeIp)
		}
	}

	return nil
}

func (s *MessageSender) localSend() error {
	var consumer java.Consumer

	lock.RLock()
	consumer = messageConsumers[s.serviceId]
	lock.RUnlock()

	if consumer != nil {
		consumer.Accept(s.msg)
	} else {
		s._bus.DealWithUnknowMessage(s.msg)
	}

	return nil
}

func (impl *CloudBusImpl3) doPublish(evt message.Event) {
	log.Glog.Debug("event", zap.String("content", dumpMessage(evt)))

	NewMessageSender(evt).Send()
}

func (impl *CloudBusImpl3) Call(msg message.NeedReplyMessage) *message.MessageReply {
	future := core.NewFutureReturnValueCompletion[message.NeedReplyMessage, *message.MessageReply](msg)

	cb := NewCloudBusCallBackImpl[*core.FutureReturnValueCompletion[message.NeedReplyMessage, *message.MessageReply]](func(reply *message.MessageReply) {
		future.Success(reply)
	}, future)

	impl.SendNeedReply(msg, cb)

	future.Wait(StaticCloudBusGlobalProperty.SyncCallTimeout)
	return future.Result
}

func (impl *CloudBusImpl3) Calls(msgs []message.NeedReplyMessage) []*message.MessageReply {
	if len(msgs) == 0 {
		panic("cannot call empty messages")
	}

	future := core.NewFutureReturnValueCompletion[message.NeedReplyMessage, []*message.MessageReply](msgs[0])

	cb := NewCloudBusListCallBackImpl[*core.FutureReturnValueCompletion[message.NeedReplyMessage, []*message.MessageReply]](func(replies []*message.MessageReply) {
		future.Success(replies)
	}, future)

	impl.SendNeedReplies(msgs, cb)

	future.Wait(StaticCloudBusGlobalProperty.SyncCallTimeout)
	return future.Result
}

func IsLogMessage(msg message.Message) bool {
	// todo api
	if StaticCloudBusGlobalProperty.MessageLogFilterAll {
		return !filterMsgNames[msg.GetClass()]
	} else {
		return filterMsgNames[msg.GetClass()]
	}
}

func (impl *CloudBusImpl3) beforeDeliverMessage(msg message.Message) {
	var interceptors []message.BeforeDeliveryMessageInterceptor
	lock.RLock()
	interceptors = beforeDeliveryMessageInterceptors[msg.GetClass()]
	lock.RUnlock()

	if len(interceptors) != 0 {
		for _, interceptor := range interceptors {
			interceptor.BeforeDeliveryMessage(msg)
		}
	}

	for _, interceptor := range beforeDeliveryMessageInterceptorsForAll {
		interceptor.BeforeDeliveryMessage(msg)
	}
}

type EndPointImpl struct {
	_bus         *CloudBusImpl3
	serv         header.Service
	registration ConsumerReceipt
	consumer     java.Consumer
	syncLevel    int
}

func (i *EndPointImpl) GetClass() string {
	return "EndPointImpl"
}

func (i *EndPointImpl) Accept(v interface{}) {
	msg, ok := v.(message.Message)
	if !ok {
		return
	}

	log.Glog.Debug(fmt.Sprintf("[msg received]: %s", dumpMessage(msg)))

	i._bus.beforeDeliverMessage(msg)
	er := i.serv.HandleMessage(msg)
	i._bus.ReplayErrorByMessageType(msg, er)
}

func (i *EndPointImpl) Active() {
	i.registration = i._bus.on(i.serv.GetId(), i.consumer)
}

func (i *EndPointImpl) Inactive() {
	if i.registration != nil {
		i.registration.Cancel()
	}
}

func (impl *CloudBusImpl3) RegisterService(serv header.Service) error {
	syncLevel := serv.GetSyncLevel()

	var endPoint EndPoint = &EndPointImpl{
		_bus:      impl,
		serv:      serv,
		syncLevel: syncLevel,
	}

	if _, ok := endPoints[serv.GetId()]; ok {
		panic(fmt.Sprintf("duplicate services[id:%s]", serv.GetId()))
	}
	endPoints[serv.GetId()] = endPoint
	endPoint.Active()

	log.Glog.Debug(fmt.Sprintf("registered service[%s]", serv.GetId()))

	return nil
}

func (impl *CloudBusImpl3) UnregisterService(serv header.Service) {
	ep, ok := endPoints[serv.GetId()]
	if !ok {
		return
	}

	ep.(EndPoint).Inactive()
	delete(endPoints, serv.GetId())
}

type EventSubscriberReceiptImpl struct {
	events []message.Event
}

func (i *EventSubscriberReceiptImpl) Unsubscribe(e message.Event) {
	lock.Lock()

	tmp := eventListeners[e.GetType().String()]
	if tmp != nil {
		delete(eventListeners, e.GetType().String())
	}

	lock.Unlock()
}

func (i *EventSubscriberReceiptImpl) UnsubscribeAll() {
	for _, event := range i.events {
		i.Unsubscribe(event)
	}
}

func (impl *CloudBusImpl3) SubscribeEvent(listener CloudBusEventListener, events ...message.Event) EventSubscriberReceipt {
	key := PlatformGetManagementServerId()

	for _, event := range events {
		lock.Lock()

		tmp := eventListeners[event.GetType().String()]
		if tmp == nil {
			tmp = make(map[string]CloudBusEventListener)
			eventListeners[event.GetType().String()] = tmp
		}
		tmp[key] = listener

		lock.Unlock()
	}

	return &EventSubscriberReceiptImpl{
		events: events,
	}
}

func (impl *CloudBusImpl3) DealWithUnknowMessage(msg message.Message) {
	details := fmt.Sprintf("No service deals with message: %s", dumpMessage(msg))

	if _, ok := msg.(message.APISyncCallMessage); ok {
		r := new(message.APIReply)
		r.SetError(errorcode.NewErr(errorcode.UNKNOWN_MESSAGE_SysErrors, details))
		r.SetSuccess(false)

		impl.Reply(msg, r)
	} else if _, ok = msg.(message.APIMessage); ok {
		evt := &message.APIEvent{
			ApiId:   msg.GetId(),
			Success: false,
		}
		evt.SetError(errorcode.NewErr(errorcode.UNKNOWN_MESSAGE_SysErrors, details))
		impl.Publish(evt)
	} else if _, ok = msg.(message.NeedReplyMessage); ok {
		r := &message.MessageReply{}
		r.SetError(errorcode.NewErr(errorcode.UNKNOWN_MESSAGE_SysErrors, details))
		r.SetSuccess(false)

		impl.Reply(msg, r)
	}

	log.Glog.Warn("Dropped an unknown message", zap.String("content", details))
}

func (impl *CloudBusImpl3) replyErrorIfMessageNeedReply(msg message.Message, er *errorcode.ErrorCode) {
	if _, ok := msg.(message.NeedReplyMessage); ok {
		reply := &message.MessageReply{}
		reply.SetError(er)
		reply.SetSuccess(false)

		impl.Reply(msg, reply)
	}
}

func (impl *CloudBusImpl3) replyErrorExistingApiEvent(evt message.APIEvent, er *errorcode.ErrorCode) {
	evt.SetError(er)
	evt.SetSuccess(false)

	impl.Publish(evt)
}

func (impl *CloudBusImpl3) replyErrorToApiMessage(msg message.APIMessage, er *errorcode.ErrorCode) {
	if _, ok := msg.(message.APISyncCallMessage); ok {
		reply := new(message.APIReply)
		reply.SetError(er)
		reply.SetSuccess(false)

		impl.Reply(msg, reply)
	} else if _, ok = msg.(message.APIMessage); ok {
		evt := &message.APIEvent{
			ApiId:   msg.GetId(),
			Success: false,
		}
		evt.SetError(er)
		impl.Publish(evt)
	}
}

func (impl *CloudBusImpl3) ReplayErrorByMessageType(msg message.Message, er *errorcode.ErrorCode) {
	if v, ok := msg.(message.APIMessage); ok {
		impl.replyErrorToApiMessage(v, er)
		return
	}
	if v, ok := msg.(message.APIEvent); ok {
		impl.replyErrorExistingApiEvent(v, er)
		return
	}

	impl.replyErrorIfMessageNeedReply(msg, er)
}

func (impl *CloudBusImpl3) toServiceId(serviceId, mgmtId string) string {
	return mgmtId + SERVICE_ID_SPLITTER + serviceId
}

func (impl *CloudBusImpl3) GetServiceId(targetServiceId string) string {
	ls := strings.Split(targetServiceId, SERVICE_ID_SPLITTER)
	return ls[1]
}

func (impl *CloudBusImpl3) MakeLocalServiceId(serviceId string) string {
	return impl.toServiceId(serviceId, PlatformGetManagementServerId())
}

func (impl *CloudBusImpl3) MakeLocalServiceIdMsg(msg message.Message, serviceId string) {
	msg.SetServiceId(impl.MakeLocalServiceId(serviceId))
}

func (impl *CloudBusImpl3) MakeServiceIdByManagementNodeId(serviceId, managementNodeId string) string {
	return impl.toServiceId(serviceId, managementNodeId)
}

func (impl *CloudBusImpl3) MakeServiceIdByManagementNodeIdMsg(msg message.Message, serviceId, managementNodeId string) {
	msg.SetServiceId(impl.MakeServiceIdByManagementNodeId(serviceId, managementNodeId))
}

func (impl *CloudBusImpl3) MakeTargetServiceIdByResourceUuid(serviceId, resourceUuid string) string {
	if serviceId == "" {
		panic("serviceId cannot be null")
	}
	if resourceUuid == "" {
		panic("resourceUuid cannot be null")
	}

	mgmtUuid := impl.DestMaker.MakeDestination(resourceUuid)
	return impl.toServiceId(serviceId, mgmtUuid)
}

func (impl *CloudBusImpl3) MakeTargetServiceIdByResourceUuidMsg(msg message.Message, serviceId, resourceUuid string) {
	msg.SetServiceId(impl.MakeTargetServiceIdByResourceUuid(serviceId, resourceUuid))
}

func (impl *CloudBusImpl3) InstallBeforeDeliveryMessageInterceptor(interceptor message.BeforeDeliveryMessageInterceptor, classes ...message.Message) {
	if len(classes) == 0 { // order bigger is mor head
		var order int
		for idx, i := range beforeDeliveryMessageInterceptorsForAll {
			if i.OrderOfBeforeDeliveryMessageInterceptor() <= interceptor.OrderOfBeforeDeliveryMessageInterceptor() {
				order = idx
				break
			}
		}

		tmp := make([]message.BeforeDeliveryMessageInterceptor, len(beforeEventPublishInterceptorsForAll)+1)
		copy(beforeDeliveryMessageInterceptorsForAll[:order], tmp)
		tmp[order] = interceptor
		copy(beforeDeliveryMessageInterceptorsForAll[order:], tmp)
		beforeDeliveryMessageInterceptorsForAll = tmp
		return
	}

	for _, clz := range classes {
		var is []message.BeforeDeliveryMessageInterceptor
		lock.RLock()
		is = beforeDeliveryMessageInterceptors[clz.GetClass()]
		lock.RUnlock()

		if is == nil {
			is = make([]message.BeforeDeliveryMessageInterceptor, 0)
		}

		lock.Lock()
		defer lock.Unlock()

		var order int
		for idx, i := range is {
			if i.OrderOfBeforeDeliveryMessageInterceptor() <= interceptor.OrderOfBeforeDeliveryMessageInterceptor() {
				order = idx
				break
			}
		}

		tmp := make([]message.BeforeDeliveryMessageInterceptor, len(is)+1)
		copy(is[:order], tmp)
		tmp[order] = interceptor
		copy(is[order:], tmp)
		beforeDeliveryMessageInterceptors[clz.GetClass()] = tmp
		return
	}
}

func (impl *CloudBusImpl3) InstallBeforeSendMessageInterceptor(interceptor message.BeforeSendMessageInterceptor, classes ...message.Message) {
	if len(classes) == 0 { // order bigger is mor head
		var order int
		for idx, i := range beforeSendMessageInterceptorsForAll {
			if i.OrderOfBeforeSendMessageInterceptor() <= interceptor.OrderOfBeforeSendMessageInterceptor() {
				order = idx
				break
			}
		}

		tmp := make([]message.BeforeSendMessageInterceptor, len(beforeEventPublishInterceptorsForAll)+1)
		copy(beforeSendMessageInterceptorsForAll[:order], tmp)
		tmp[order] = interceptor
		copy(beforeSendMessageInterceptorsForAll[order:], tmp)
		beforeSendMessageInterceptorsForAll = tmp
		return
	}

	for _, clz := range classes {
		var is []message.BeforeSendMessageInterceptor
		lock.RLock()
		is = beforeSendMessageInterceptors[clz.GetClass()]
		lock.RUnlock()

		if is == nil {
			is = make([]message.BeforeSendMessageInterceptor, 0)
		}

		lock.Lock()
		defer lock.Unlock()

		var order int
		for idx, i := range is {
			if i.OrderOfBeforeSendMessageInterceptor() <= interceptor.OrderOfBeforeSendMessageInterceptor() {
				order = idx
				break
			}
		}

		tmp := make([]message.BeforeSendMessageInterceptor, len(is)+1)
		copy(is[:order], tmp)
		tmp[order] = interceptor
		copy(is[order:], tmp)
		beforeSendMessageInterceptors[clz.GetClass()] = tmp
		return
	}
}

// len(classes)==0, set global BeforePublishEventInterceptor, or set BeforePublishEventInterceptor for one event
func (impl *CloudBusImpl3) InstallBeforePublishEventInterceptor(interceptor message.BeforePublishEventInterceptor, classes ...message.Event) {
	if len(classes) == 0 { // order bigger is mor head
		var order int
		for idx, i := range beforeEventPublishInterceptorsForAll {
			if i.OrderOfBeforePublishEventInterceptor() <= interceptor.OrderOfBeforePublishEventInterceptor() {
				order = idx
				break
			}
		}

		tmp := make([]message.BeforePublishEventInterceptor, len(beforeEventPublishInterceptorsForAll)+1)
		copy(beforeEventPublishInterceptorsForAll[:order], tmp)
		tmp[order] = interceptor
		copy(beforeEventPublishInterceptorsForAll[order:], tmp)
		beforeEventPublishInterceptorsForAll = tmp
		return
	}

	for _, clz := range classes {
		var is []message.BeforePublishEventInterceptor
		lock.RLock()
		is = beforeEventPublishInterceptors[clz.GetClass()]
		lock.RUnlock()

		if is == nil {
			is = make([]message.BeforePublishEventInterceptor, 0)
		}

		lock.Lock()
		defer lock.Unlock()

		var order int
		for idx, i := range is {
			if i.OrderOfBeforePublishEventInterceptor() <= interceptor.OrderOfBeforePublishEventInterceptor() {
				order = idx
				break
			}
		}

		tmp := make([]message.BeforePublishEventInterceptor, len(is)+1)
		copy(is[:order], tmp)
		tmp[order] = interceptor
		copy(is[order:], tmp)
		beforeEventPublishInterceptors[clz.GetClass()] = tmp
		return
	}
}

func (impl *CloudBusImpl3) populateExtension() {

}

func (impl *CloudBusImpl3) Start() error {
	SERVICE_ID = impl.MakeLocalServiceId("cloudbus.messages")
	EVENT_ID = impl.MakeLocalServiceId("cloudbus.events")

	impl.on(SERVICE_ID, new(messageConsumer))
	impl.on(EVENT_ID, new(eventConsumer))

	impl.populateExtension()

	return nil
}

func (impl *CloudBusImpl3) Stop() error {
	return nil
}

func (impl *CloudBusImpl3) doSendAndCallExtensions(msg message.Message) {
	// for unit test finding invocation chain
	// MessageCommandRecorder.record(msg.GetClass()) // todo

	var interceptors []message.BeforeSendMessageInterceptor
	lock.RLock()
	interceptors = beforeSendMessageInterceptors[msg.GetClass()]
	lock.RUnlock()

	if len(interceptors) != 0 {
		for _, interceptor := range interceptors {
			interceptor.BeforeSendMessage(msg)
		}
	}

	for _, interceptor := range beforeSendMessageInterceptorsForAll {
		interceptor.BeforeSendMessage(msg)
	}

	impl.doSend(msg)
}

func (impl *CloudBusImpl3) doSend(msg message.Message) {
	// evalThreadContextToMessage(msg) // todo :当前的thread Context中获取必要信息。每个api调用所携带的uuid就是这样传递下去的
	log.Glog.Debug(fmt.Sprintf("[msg send]: %s", dumpMessage(msg)))

	NewMessageSender(msg).Send()
}

func (impl *CloudBusImpl3) send(msg message.Message, noNeedReply bool) {
	if msg.GetServiceId() == "" {
		panic(fmt.Sprintf("service id cannot be null: %s", msg.GetClass()))
	}

	msg.PutHeaderEntry(CORRELATION_ID, msg.GetId())
	msg.PutHeaderEntry(REPLY_TO, SERVICE_ID) // 消息的回复队列设置
	if msg.GetTypeName() == message.TypeAPIMessage {
		// api always need reply
		msg.PutHeaderEntry(NO_NEED_REPLY_MSG, "false")
	} else if msg.GetTypeName() == message.TypeNeedReplyMessage {
		// for NeedReplyMessage sent without requiring receiver to reply,
		// mark it, then it will not be tracked and replied
		msg.PutHeaderEntry(NO_NEED_REPLY_MSG, fmt.Sprintf("%t", noNeedReply))
	}

	impl.doSendAndCallExtensions(msg)
}
